﻿using System;
using System.Collections.Generic;
using System.Text;
//new
using Apache.NMS;
using Apache.NMS.ActiveMQ;
using System.Threading;
using System.Xml;

namespace DataRecv.models.mq
{
    class MQ
    {
        private string m_uri;
        private string m_SendTopic;
        private IConnectionFactory m_factory;
        private IConnection m_connection;
        private ISession m_session;
        private IMessageProducer m_producer;

        public delegate void Log(string txt);
        public delegate void MessageInComing(string txt);
        public event Log LogEvent;
        public bool link;//连接状态

        public MQ()
        {
            m_producer = null;
            m_factory = null;
            m_connection = null;
            m_session = null;
        }

        public MQ(string mquri, string mqSendTopic)
        {
            m_uri = mquri;
            m_SendTopic = mqSendTopic;
            link = false;
        }

        ~MQ()
        {
            Stop();
        }

        public void Start()
        {
            try
            {
                m_factory = new ConnectionFactory(m_uri);
                m_connection = m_factory.CreateConnection();
                if (m_connection != null)
                {
                    m_connection.ConnectionInterruptedListener += new ConnectionInterruptedListener(connection_ConnectionInterruptedListener);
                    m_connection.ConnectionResumedListener += new ConnectionResumedListener(connection_ConnectionResumedListener);
                    m_connection.ExceptionListener += new ExceptionListener(connection_ExceptionListener);
                    m_connection.Start();
                    Resume();
                }
            }
            catch (Exception er)
            {
                if (LogEvent != null) { LogEvent("MQ Start()异常" + er.Message); }
            }
        }

        public void Resume()
        {
            try
            {
                m_session = m_connection.CreateSession();
                if (m_session != null)
                {
                    m_producer = m_session.CreateProducer(new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic(m_SendTopic));
                    if ((m_producer != null) && (m_connection.IsStarted))
                    {
                        link = true;
                        if (LogEvent != null) { LogEvent("MQ启动"); }
                    }
                }
            }
            catch (Exception er)
            {
                if (LogEvent != null) { LogEvent("MQ Resume()异常" + er.Message); }
            }
        }

        public void Stop()
        {
            try
            {
                if (m_producer != null)
                {
                    m_producer.Close();
                    m_producer.Dispose();
                    m_producer = null;
                }
            }
            catch { }

            try
            {
                if (m_session != null)
                {
                    m_session.Close();
                    m_session.Dispose();
                    m_session = null;
                }
            }
            catch { }

            link = false;
            if (LogEvent != null) { LogEvent("MQ停止"); }
        }

        public void Close()
        {
            Stop();
            try
            {
                if (m_connection != null)
                {
                    m_connection.ConnectionResumedListener -= connection_ConnectionResumedListener;
                    m_connection.ExceptionListener -= connection_ExceptionListener;
                    m_connection.ConnectionInterruptedListener -= connection_ConnectionInterruptedListener;
                    m_connection.Stop();
                    m_connection.Close();
                    m_connection.Dispose();
                    m_connection = null;
                }
            }
            catch
            { }
            if (LogEvent != null) { LogEvent("MQ退出"); }
        }

        void connection_ExceptionListener(Exception exception)
        {
            if (LogEvent != null) { LogEvent("connection_ExceptionListener()异常" + exception.Message); }
        }

        void connection_ConnectionResumedListener()
        {
            if (LogEvent != null) { LogEvent("MQ重连"); }
            Resume();
        }

        void connection_ConnectionInterruptedListener()
        {
            if (LogEvent != null) { LogEvent("MQ断开"); }
            Stop();
        }



        public bool SendMQMessage(string strText, string key, string value)
        {
            bool ans = false;
            try
            {
                if (m_producer != null)
                {
                    ITextMessage msg = m_producer.CreateTextMessage();
                    msg.Text = strText;
                    msg.Properties.Clear();
                    msg.Properties.SetString(key, value);
                    if ((m_connection != null) && (m_connection.IsStarted))
                    {
                        m_producer.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);
                        ans = true;
                    }
                }
            }
            catch (Exception er)
            {
                ans = false;
                if (LogEvent != null) { LogEvent("SendMQMessage()异常" + er.Message); }
                //if (er.Message.Contains("The connection is already closed"))
                //{
                //    link = false;
                //    Resume();
                //}
            }
            return ans;
        }


        public bool SendMQMessage(string strText)
        {
            bool ans = false;
            try
            {
                if (m_producer != null)
                {
                    ITextMessage msg = m_producer.CreateTextMessage();
                    msg.Text = strText;
                    if ((m_connection != null) && (m_connection.IsStarted))
                    {
                        m_producer.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);
                        ans = true;
                    }
                }
            }
            catch (Exception er)
            {
                ans = false;
                if (LogEvent != null) { LogEvent("SendMQMessage()异常" + er.Message); }
            }
            return ans;
        }

        private void CreateNode(XmlDocument xmldoc, string parentnodename, string nodename, string innertext, Dictionary<string, string> attributes)
        {
            if ((xmldoc != null) && (parentnodename != null) && (parentnodename != "") && (nodename != null) && (nodename != ""))
            {
                XmlNode parentnode = xmldoc.SelectSingleNode("descendant::" + parentnodename);
                if (parentnode != null)
                {
                    XmlElement xe = xmldoc.CreateElement(nodename);
                    if (innertext != null)
                    {
                        xe.InnerText = innertext;
                    }
                    if (attributes != null)
                    {
                        foreach (KeyValuePair<string, string> pair in attributes)
                        {
                            xe.SetAttribute(pair.Key, pair.Value);
                        }
                    }
                    parentnode.AppendChild(xe);
                }
            }
        }

    }
}
